package com.atguigu.flink.sql.function;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.ScalarFunction;

/**
 * Created by Smexy on 2023/4/14
 *
 *  聚合函数 将多行数据里的标量值转换成一个新标量值
 *          N行N列进，一行一列出的函数。
 *
 *   求每种传感器的平均水位
 *
 *   ①按照flink的规范，定义一个函数类
 *   ②在系统中注册函数，提供一个函数名字
 *   ③可以调用函数
 */
public class Demo4_CustomUDAFFunciton
{
    public static void main(String[] args) {

        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tableEnvironment = TableEnvironment.create(environmentSettings);

        String createTableSql = " create table t1 ( id STRING, ts BIGINT , vc INT ," +
            "  `file.path` STRING NOT NULL METADATA " +
            "    )WITH (" +
            "  'connector' = 'filesystem'," +
            "  'path' = 'data/sensor.txt'," +
            "  'format' = 'csv'" +
            ") ";

        tableEnvironment.executeSql(createTableSql);

        //②在系统中注册函数，提供一个函数名字
        MyAvg myFunction = new MyAvg();
        tableEnvironment.createTemporaryFunction("myAvg",myFunction);

        tableEnvironment.sqlQuery(" select id,myAvg(vc) avgVc from t1  group by id").execute().print();

    }

    /*
        自定义聚合函数是通过扩展 AggregateFunction<T 返回的类型, ACC> 来实现的。
        下面几个方法是每个 AggregateFunction 必须要实现的：

                createAccumulator()
                accumulate()
                getValue()
     */

    public static class MyAvg extends AggregateFunction<Double,MyAcc>
    {

        /*
            第一个参数是累加器，后续的参数是传入的要聚合的列
            实现聚合逻辑
         */
        public void accumulate(MyAcc acc, Integer vc) {
          acc.num += 1;
          acc.sumVc += vc;
        }


        //返回最终结果
        @Override
        public Double getValue(MyAcc accumulator) {
            return accumulator.sumVc / accumulator.num;
        }

        //创建累加器对象
        @Override
        public MyAcc createAccumulator() {
            return new MyAcc(0,0d);
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyAcc{

        private Integer num;
        private Double sumVc;
     }
}
